package com.mq.produce.simple.v2;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;

/**
 * @Classname CustomerProduceParameters
 * @Description 提高生产者吞吐量
 * @Version 1.0.0
 * @Date 2025/3/11 17:52
 * @Created by ncs
 */
public class CustomerProduceParameters {

    static String kafkaServer = "192.168.125.100:9093,192.168.125.100:9094,192.168.125.100:9095";

    public static void main(String[] args) throws Exception{
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //1. 提高缓冲区大小 默认32m 针对所有分区的缓存大小 提升到64m
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);
        //2. 批次大小 默认16k,针对一个分区每次数据满多少发送 提升到32k
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 327680);
        //3. linger.ms  每隔多少时间发送 默认0 代表即时发送
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 10000);
        //4. 压缩方式 可配置值gzip,snappy,lz4,zstd,一般用snappy
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        for (int i = 0; i < 1; i++) {
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("first-topic","first-message-callback" + i);
            System.out.println("开始时间" + sdf.format(new Date()));
            kafkaProducer.send(producerRecord, (metadata, exception) -> {
                SimpleDateFormat t = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                if (exception != null) {
                    System.out.println("发送消息异步回调失败,异常信息=" + exception);
                }else {
                    System.out.println(t.format(new Date())+"发送消息异步回调成功,topic="+ metadata.topic()+", partition="+metadata.partition());
                }
            });
            System.out.println("结束时间" + sdf.format(new Date()));
        }
        //该方法会导致消息即时发送
        kafkaProducer.close();
    }
}
